package org.cy.websocket.stomp.integration;

import okhttp3.*;
import okio.ByteString;
import org.cy.websocket.stomp.StompHeaders;
import org.cy.websocket.stomp.StompWebSocketListener;
import org.cy.websocket.stomp.messaging.Message;

/**
 * Created in 2021-05-09 16:51
 *
 * @author cao.yong
 */
public class OkHttpWebSocket extends AbstractWebSocket {

    private WebSocket baseSocket;

    @Override
    public void connect(String url, StompHeaders headers, StompWebSocketListener listener) {
        Request request = new Request.Builder().get().url(url).build();
        this.baseSocket = new OkHttpClient.Builder().build().newWebSocket(request, new WebSocketListener() {
            @Override
            public void onOpen(WebSocket webSocket, Response response) {
                listener.onOpen();
            }

            @Override
            public void onMessage(WebSocket webSocket, String text) {
                listener.onMessage(decodeMessage(text, false));
            }

            @Override
            public void onMessage(WebSocket webSocket, ByteString bytes) {
                listener.onMessage(decodeMessage(bytes.toByteArray(), true));
            }

            @Override
            public void onClosing(WebSocket webSocket, int code, String reason) {
                webSocket.close(code, reason);
            }

            @Override
            public void onClosed(WebSocket webSocket, int code, String reason) {
                listener.onClosed(code, reason);
            }

            @Override
            public void onFailure(WebSocket webSocket, Throwable ex, Response response) {
                listener.onFailure(ex);
            }
        });
    }

    @Override
    public void send(Message<byte[]> message) {
        byte[] bytes = encodeMessage(message);
        this.baseSocket.send(ByteString.of(bytes));
    }

    @Override
    public void close() {
        this.baseSocket.close(1000, "");
    }
}
